package com.jeffse.light.stream.core.data.operator;

import com.jeffse.light.stream.core.constans.CommonConstans;
import com.jeffse.light.stream.core.constans.TransformConstans;
import com.jeffse.light.stream.core.data.context.Context;
import com.jeffse.light.stream.core.data.context.OperatorContext;
import com.jeffse.light.stream.core.data.descriptor.operatorDescriptor.OperatorDescriptor;
import com.jeffse.light.stream.core.data.operator.Covert.Convert;
import com.jeffse.light.stream.core.data.stream.DataStream;
import com.jeffse.light.stream.core.function.advance.SlideWindowFunction;
import com.jeffse.light.stream.core.function.advance.WindowFunction;
import com.jeffse.light.stream.core.pojo.window.SlideWindowInfo;
import com.jeffse.light.stream.core.pojo.window.WindowInfo;
import com.jeffse.light.stream.core.function.advance.FlatMapFunction;
import com.jeffse.light.stream.core.function.advance.MapFunction;
import com.jeffse.light.stream.core.function.base.*;
import com.jeffse.light.stream.core.pojo.OutTag;
import com.jeffse.light.stream.core.selector.KeyedSelector;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

@Slf4j
public class OutPutStreamOperator<KEY, T, O>  implements StreamOperator<KEY, T, O>, OperatorContext<KEY, T, O> {

    private OutTag outTag = null;

    private OutTag sinkTag = null;

    /**
     * 上下文
     */
    Context ctx = null;

    List<OperatorDescriptor> operatorDescriptorList = new ArrayList<>();

    /**
     * 记录流程中目前的 keyedBy 的key
     */
    KeyedSelector key = null;

    /**
     * 记录流程中目前的 waterMark 的 function
     */
    WaterMarkFunction waterMark = null;

    public OutPutStreamOperator(Context ctx){
        this.ctx = ctx;
    }

    @Override
    public OutTag getOutTag() {
        return outTag;
    }

    @Override
    public void setOutTag(OutTag outTag) {
        this.outTag = outTag;
    }

    public void setSinkTag(OutTag sinkTag) {
        this.sinkTag = sinkTag;
    }

    public OutTag getSinkTag() {
        return sinkTag;
    }

    @Override
    public OutPutStreamOperator processElement(Object value, OperatorDescriptor operatorDescriptor) throws Exception {
        if(operatorDescriptor != null){
            Convert.getINSTANCE().handleData(operatorDescriptor, value);
        }else if(sinkTag != null){
            ctx.collect(value, sinkTag);
        }else {
            ctx.doSink(value);
        }
        return this;
    }

    @Override
    public OutPutStreamOperator process(WindowFunction windowFunction) throws Exception {
        OperatorDescriptor operatorDescriptor = null;
        KeyedProcessFunction keyedProcessFunction =  null;

        if(operatorDescriptorList.size() > 0){
            operatorDescriptor = operatorDescriptorList.get(operatorDescriptorList.size() -1);
            assert operatorDescriptor.getName().equals(TransformConstans.WINDOW) : CommonConstans.USE_WINDOW_FIRST;
            operatorDescriptor.setName(TransformConstans.KEYED_PROCESS);
            operatorDescriptor.setCtx(ctx);
            operatorDescriptor.setOutTag(outTag);
            WindowInfo windowInfo = operatorDescriptor.getWindowInfo();
            if(windowInfo.getType().equals(OperatorContext.WindowType.TIME)){
                keyedProcessFunction = new TimeWindowKeyedProcessFunction(windowInfo, windowFunction);
            }else{
                keyedProcessFunction = new CountWindowKeyedProcessFunction(windowInfo, windowFunction);
            }

            keyedProcessFunction.setCtx(operatorDescriptor.getCtx());
            keyedProcessFunction.setOutTag(operatorDescriptor.getOutTag());
            operatorDescriptor.setFunc(keyedProcessFunction);
        }else{
            throw new Exception(CommonConstans.USE_WINDOW_FIRST);
        }
        setNextOperatorDescriptor(operatorDescriptor);
        return this;
    }

    @Override
    public OutPutStreamOperator process(SlideWindowFunction slideWindowFunction) throws Exception {
        OperatorDescriptor operatorDescriptor = null;
        KeyedProcessFunction keyedProcessFunction =  null;
        if(operatorDescriptorList.size() > 0){
            operatorDescriptor = operatorDescriptorList.get(operatorDescriptorList.size() -1);
            assert operatorDescriptor.getName().equals(TransformConstans.SLIDE_WINDOW) : CommonConstans.USE_SLIDEWINDOW_FIRST;
            operatorDescriptor.setName(TransformConstans.KEYED_PROCESS);
            operatorDescriptor.setCtx(ctx);
            operatorDescriptor.setOutTag(outTag);

            SlideWindowInfo windowInfo = operatorDescriptor.getSlidWindowInfo();
            if(windowInfo.getType().equals(OperatorContext.WindowType.TIME)){
                keyedProcessFunction = new TimeSlideWindowKeyedProcessFunction(windowInfo, slideWindowFunction, operatorDescriptor.getWaterMarkFunction());
            }else{
                keyedProcessFunction = new CountSlideWindowKeyedProcessFunction(windowInfo, slideWindowFunction, operatorDescriptor.getWaterMarkFunction());
            }

            keyedProcessFunction.setCtx(operatorDescriptor.getCtx());
            keyedProcessFunction.setOutTag(operatorDescriptor.getOutTag());
            operatorDescriptor.setFunc(keyedProcessFunction);
        }else{
            throw new Exception(CommonConstans.USE_SLIDEWINDOW_FIRST);
        }
        setNextOperatorDescriptor(operatorDescriptor);
        return this;
    }

    /**
     * 添加过程函数
     * @param abstractProcessFunction
     * @return
     */
    @Override
    public OutPutStreamOperator process(AbstractProcessFunction abstractProcessFunction) throws Exception {
        OperatorDescriptor operatorDescriptor = null;
        if(operatorDescriptorList.size() > 0){
            operatorDescriptor = operatorDescriptorList.get(operatorDescriptorList.size() -1);
        }

        // keyby, window, slideWindow, watermark 都是 processFunction 描述符的属性
        if(operatorDescriptorList.size() > 0 && operatorDescriptor.getName().equals(TransformConstans.KEYEDBY)){
            assert abstractProcessFunction instanceof KeyedProcessFunction: CommonConstans.NOT_KEYED_PROCESSFUNCTION;
            operatorDescriptor.setFunc(abstractProcessFunction);
            operatorDescriptor.setName(TransformConstans.KEYED_PROCESS);
            operatorDescriptor.setWaterMarkFunction(waterMark);
        }else{
            if(abstractProcessFunction instanceof KeyedProcessFunction){
                throw new Exception(CommonConstans.USE_KEYED_BY_FIRST);
            }else if(abstractProcessFunction instanceof ProcessFunction){
                operatorDescriptor = new OperatorDescriptor(ctx, TransformConstans.PROCESS, abstractProcessFunction, outTag);
            }

            operatorDescriptorList.add(operatorDescriptor);
        }
        abstractProcessFunction.setCtx(ctx);
        abstractProcessFunction.setOutTag(outTag);
        setNextOperatorDescriptor(operatorDescriptor);

        return this;
    }

    @Override
    public OutPutStreamOperator map(MapFunction mapFunction) {
        OperatorDescriptor operatorDescriptor = new OperatorDescriptor(ctx, TransformConstans.MAP, mapFunction, key, outTag);
        operatorDescriptorList.add(operatorDescriptor);
        setNextOperatorDescriptor(operatorDescriptor);
        return this;
    }

    @Override
    public OutPutStreamOperator flatMap(FlatMapFunction flatMapFunction){
        OperatorDescriptor operatorDescriptor = new OperatorDescriptor(ctx, TransformConstans.FLAT_MAP, flatMapFunction, key, outTag);
        operatorDescriptorList.add(operatorDescriptor);
        setNextOperatorDescriptor(operatorDescriptor);
        return this;
    }

    @Override
    public OutPutStreamOperator union(OutTag outTag) {
        OutTag unionTag = new OutTag(outTag.getName() + "_" + this.outTag.getName());
        OutPutStreamOperator outPutStreamOperator = new OutPutStreamOperator(ctx);
        outPutStreamOperator.setOutTag(unionTag);
        ((DataStream)ctx).union(outTag, this.outTag, unionTag, outPutStreamOperator);
        return outPutStreamOperator;
    }

    @Override
    public OperatorContext window(Enum<WindowType> windowType, Integer num, Enum<UnitType> unit, Boolean isEventTime) throws Exception {
        WindowInfo windowInfo = new WindowInfo(windowType, num, unit, isEventTime);
        if(operatorDescriptorList.size() <= 0 || !operatorDescriptorList.get(operatorDescriptorList.size() -1).getName().equals(TransformConstans.KEYEDBY)){
            throw new Exception(CommonConstans.USE_KEYED_BY_FIRST);
        }

        OperatorDescriptor operatorDescriptor = operatorDescriptorList.get(operatorDescriptorList.size() -1);
        operatorDescriptor.setWindowInfo(windowInfo);
        operatorDescriptor.setName(TransformConstans.WINDOW);
        operatorDescriptor.setWaterMarkFunction(waterMark);
        return this;
    }

    @Override
    public OperatorContext slideWindow(Enum<WindowType> windowType, Integer x, Enum<UnitType> unitX, Integer y, Enum<UnitType> unitY, Boolean isEventTime) throws Exception {
        SlideWindowInfo windowInfo = new SlideWindowInfo(windowType, x, unitX, y, unitY, isEventTime);
        if(operatorDescriptorList.size() <= 0 || !operatorDescriptorList.get(operatorDescriptorList.size() -1).getName().equals(TransformConstans.KEYEDBY)){
            throw new Exception(CommonConstans.USE_KEYED_BY_FIRST);
        }
        OperatorDescriptor operatorDescriptor = operatorDescriptorList.get(operatorDescriptorList.size() -1);
        operatorDescriptor.setSlidWindowInfo(windowInfo);
        operatorDescriptor.setName(TransformConstans.SLIDE_WINDOW);
        operatorDescriptor.setWaterMarkFunction(waterMark);
        return this;
    }

    @Override
    public OutPutStreamOperator keyedBy(KeyedSelector keyedSelector) {
        key = keyedSelector;
        operatorDescriptorList.add(new OperatorDescriptor(ctx, TransformConstans.KEYEDBY, null, keyedSelector));
        return this;
    }

    @Override
    public StreamOperator transform(String functionType) {
        return this;
    }

    @Override
    public OutPutStreamOperator setWaterMark(WaterMarkFunction waterMarkFunction) {
        waterMark = waterMarkFunction;
        return this;
    }

    private void setNextOperatorDescriptor(OperatorDescriptor operatorDescriptor){
        if(operatorDescriptorList.size() > 1){
            operatorDescriptorList.get(operatorDescriptorList.size() - 2).setNetOperatorDescriptor(operatorDescriptor);
        }
    }

    @Override
    public OperatorDescriptor getFirstDescriptor(){
        if(operatorDescriptorList.size() > 0){
            return operatorDescriptorList.get(0);
        }
        return null;
    }

    @Override
    public Integer count() {
        return null;
    }

    @Override
    public Double avg() {
        return null;
    }

    @Override
    public Double sum() {
        return null;
    }

    @Override
    public Double max() {
        return null;
    }

    @Override
    public Double min() {
        return null;
    }
}
